package kafka

import (
	"context"
	"encoding/json"
	"fmt"

	"gitee.com/go-course/go11/devcloud-mini/maudit/apps/log"
	"gitee.com/go-course/go11/devcloud-mini/maudit/conf"
	"gitee.com/go-course/go11/devcloud-mini/maudit/exporter"
	"github.com/infraboard/mcube/ioc"
	"github.com/segmentio/kafka-go"
)

func NewKafkaOpeateLogProvider() *KafkaOpeateLogProvider {
	kc := conf.C().Kafka
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  kc.Brokers,
		GroupID:  kc.GroupId,
		Topic:    kc.Topic,
		MaxBytes: 10e6, // 10MB
	})

	return &KafkaOpeateLogProvider{
		reader: r,
		svc:    ioc.GetController(log.AppName).(log.Service),
	}
}

// 由kafka来提供操作日志数据
// 1. 读取 操作日志发送的topic里面的数据
// 2. 调用save接口进行保持
// 运行起来后，一直跑在后台的, 有点类似于一个agent
// 可以理解为一个常驻在后台进行独立任务处理的goroutine
type KafkaOpeateLogProvider struct {
	// kafka reader
	reader *kafka.Reader
	//
	svc log.Service
}

// 1. 读取 操作日志发送的topic里面的数据
// 2. 调用save接口进行保持
func (k *KafkaOpeateLogProvider) Run(ctx context.Context) {
	// 函数退出关闭reader
	defer func() {
		if err := k.reader.Close(); err != nil {
			fmt.Println("failed to close reader:", err)
		}
	}()

	fmt.Println("start consumer operate log")
	for {
		// 读取数据
		m, err := k.reader.ReadMessage(ctx)
		// 1. 记录程序获取的消息的数量
		// 2. 记录成功处理的消息的数量
		// 3. 记录没被成功处理的消息的数据
		exporter.C.OperateLogCountInc()
		if err != nil {
			fmt.Println(err)
			continue
		}
		// []byte ---> operatelog obj
		record := &log.OperateLog{}
		err = json.Unmarshal(m.Value, record)
		if err != nil {
			// 尽量不要把数据一直挤压在 kafka里面
			fmt.Println(err)
			continue
		}
		_, err = k.svc.Save(ctx, record)
		if err != nil {
			fmt.Println(err)
			continue
		}
	}

}
